package com.ams.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ams.io.BufferAllocator;
import com.ams.io.BufferFactory;
import com.ams.io.network.NetworkHandler;
import com.ams.io.network.SocketAcceptor;
import com.ams.io.network.connection.ConnectionListener;
import com.ams.io.network.connection.Connection;
import com.ams.io.network.connection.NetworkConnector;
import com.ams.server.handler.HttpHandler;
import com.ams.server.handler.ProtocolHandler;
import com.ams.server.handler.RtmpHandler;
import com.ams.server.handler.RtmpTunnelHandler;
import com.ams.server.replication.ReplPublishHandler;
import com.ams.server.replication.ReplSubscribeHandler;

public class Server implements ConnectionListener {
    private Logger logger = LoggerFactory.getLogger(Server.class);

    private class WorkerThreadPoolExecutor extends ThreadPoolExecutor {
        private static final int CORE_POOL_SIZE = 8;
        private static final int QUEUE_SIZE = 256;

        public WorkerThreadPoolExecutor(int maxPoolSize) {
            super(CORE_POOL_SIZE, maxPoolSize, 60, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<Runnable>(QUEUE_SIZE));
        }

        protected void afterExecute(Runnable r, Throwable t) {
            if (r instanceof ProtocolHandler) {
                ProtocolHandler handler = (ProtocolHandler) r;
                if (handler.isKeepAlive()) {
                    execute(handler);
                }
            }
        }
    }

    private Configuration config;
    private HashMap<SocketAddress, NetworkHandler> acceptorMap;
    private HashMap<SocketAddress, ProtocolHandler> handlerMap;
    private ArrayList<ProtocolHandler> replicationSubscribeHandlers;

    private WorkerThreadPoolExecutor executor;

    public Server(Configuration config) throws IOException {
        initByteBufferFactory(config);
        this.config = config;
        this.acceptorMap = new HashMap<SocketAddress, NetworkHandler>();
        this.handlerMap = new HashMap<SocketAddress, ProtocolHandler>();
        this.replicationSubscribeHandlers = new ArrayList<ProtocolHandler>();

        int poolSize = config.getWokerThreadPoolSize();
        executor = new WorkerThreadPoolExecutor(poolSize);

        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                shutdown();
            }
        });
    }

    private void initByteBufferFactory(Configuration config) {
        BufferAllocator allocator = new BufferAllocator();
        if (config.getMemPoolSize() > 0) {
            allocator.setPoolSize(config.getMemPoolSize());
        }
        allocator.init();
        BufferFactory.setAllocator(allocator);
    }

    public void addTcpListenEndpoint(SocketAddress endpoint,
            ProtocolHandler handler) throws IOException {
        int dispatcherSize = config.getDispatcherThreadPoolSize();
        SocketAcceptor acceptor = new SocketAcceptor(endpoint, dispatcherSize);
        acceptor.setConnectionListener(this);
        acceptor.setSocketProperties(config.getSocketProperties());
        acceptorMap.put(endpoint, acceptor);
        handlerMap.put(endpoint, handler);
    }

    private void createServiceHandler() {
        try {
            // http service
            if (config.getHttpHost() != null) {
                InetSocketAddress httpEndpoint = new InetSocketAddress(
                        config.getHttpHost(), config.getHttpPort());
                addTcpListenEndpoint(httpEndpoint,
                        new HttpHandler(config.getHttpContextRoot()));
            }

        } catch (Exception e) {
            logger.info("Creating http service failed.");
        }

        // normal rtmp service
        if (config.getRtmpHost() != null) {
            try {
                InetSocketAddress rtmpEndpoint = new InetSocketAddress(
                        config.getRtmpHost(), config.getRtmpPort());
                addTcpListenEndpoint(rtmpEndpoint,
                        new RtmpHandler(config.getRtmpContextRoot()));
            } catch (Exception e) {
                logger.info("Creating rtmp service failed.");
            }

            try {
                // rtmpt service
                InetSocketAddress rtmptEndpoint = new InetSocketAddress(
                        config.getRtmpHost(), config.getRtmptPort());
                RtmpTunnelHandler handler = new RtmpTunnelHandler(
                        config.getRtmpContextRoot());
                handler.setExecutor(executor);
                addTcpListenEndpoint(rtmptEndpoint, handler);
            } catch (Exception e) {
                logger.info("Creating rtmpt service failed.");
            }
        }

    }

    public void createReplicationHandler() {
        // create replication publisher handler
        if (config.getReplicationRtmpHost() != null) {
            try {
                InetSocketAddress replicationEndpoint = new InetSocketAddress(
                        config.getReplicationRtmpHost(),
                        config.getReplicationRtmpPort());
                addTcpListenEndpoint(replicationEndpoint,
                        new ReplPublishHandler());
            } catch (Exception e) {
                logger.info("Creating replication rtmp service failed.");
            }
        }

        // create replication subscriber handlers
        if (config.getReplicationPublisherHosts() != null) {
            int publisherPort = config.getReplicationPublisherPort();
            for (String publisherHost : config.getReplicationPublisherHosts()) {
                ProtocolHandler handler = new ReplSubscribeHandler(
                        publisherHost, publisherPort);
                replicationSubscribeHandlers.add(handler);
            }
        }

    }

    private void startupService() {
        for (SocketAddress endpoint : acceptorMap.keySet()) {
            NetworkHandler acceptor = acceptorMap.get(endpoint);
            acceptor.start("acceptor");
            logger.info("Start service on port: {}", endpoint);
        }
        for (ProtocolHandler handler : replicationSubscribeHandlers) {
            executor.execute(handler);
        }
    }

    public void startup() {
        createServiceHandler();
        createReplicationHandler();
        startupService();
        logger.info("Server is started.");
    }

    public void shutdown() {
        for (ProtocolHandler handler : replicationSubscribeHandlers) {
            handler.close();
        }

        for (SocketAddress endpoint : acceptorMap.keySet()) {
            NetworkHandler acceptor = acceptorMap.get(endpoint);
            acceptor.stop();
            logger.info("Shutdown service on port: {}", endpoint);
            System.out.println("Shutdown service on port: " + endpoint);
        }

        executor.shutdown();

        logger.info("Server is shutdown.");
        System.out.println("Server is shutdown.");
    }

    public ProtocolHandler getProtocolHandler(SocketAddress endpoint) {
        if (handlerMap.containsKey(endpoint)) {
            return handlerMap.get(endpoint);
        } else {
            if (endpoint instanceof InetSocketAddress) {
                SocketAddress endpointAny = new InetSocketAddress("0.0.0.0",
                        ((InetSocketAddress) endpoint).getPort());
                if (handlerMap.containsKey(endpointAny)) {
                    return handlerMap.get(endpointAny);
                }
            }
        }
        return null;
    }

    protected void launchNewHandler(ProtocolHandler handler, Connection conn) {
        ProtocolHandler newHandler = handler.handlerFactory();
        newHandler.setConnection(conn);
        newHandler.setExecutor(executor);
        executor.execute(newHandler);
    }

    public void connectionEstablished(Connection conn) {
        NetworkConnector networkConn = (NetworkConnector) conn;
        SocketAddress endpoint = networkConn.getLocalEndpoint();
        ProtocolHandler protocolHandler = getProtocolHandler(endpoint);
        if (protocolHandler != null) {
            launchNewHandler(protocolHandler, conn);
        } else {
            logger.warn("unkown connection port:{}", endpoint);
        }
    }

    public void connectionClosed(Connection conn) {
    }

}